#include "RpcProvider.h"
#include "MprpcApplication.h"
#include "rpcheader.pb.h"
#include "ZooKeeperUtil.h"
#include "MprpcController.h"
#include <string>
#include <iostream>

class RpcDone : public google::protobuf::Closure
{
public:
    RpcDone(const muduo::net::TcpConnectionPtr &conn,
            std::unique_ptr<google::protobuf::Message> request,
            std::unique_ptr<google::protobuf::Message> response)
        : conn_(conn),
          request_(std::move(request)),
          response_(std::move(response)) {}

    void Run() override
    {
        std::string response_str;
        if (response_->SerializeToString(&response_str))
        {
            conn_->send(response_str);
        }
        else
        {
            std::cerr << "response serialize failed!" << std::endl;
        }
        delete this; // 释放自身（Closure 一般都是这么用）
    }

    // 提供裸指针给 CallMethod 使用
    google::protobuf::Message *GetRequest() { return request_.get(); }
    google::protobuf::Message *GetResponse() { return response_.get(); }

private:
    muduo::net::TcpConnectionPtr conn_;
    std::unique_ptr<google::protobuf::Message> request_;
    std::unique_ptr<google::protobuf::Message> response_;
};

void RpcProvider::NotifyService(google::protobuf::Service *service)
{
    ServiceInfo service_info;

    const google::protobuf::ServiceDescriptor *pserviceDesc = service->GetDescriptor();
    std::string service_name = pserviceDesc->name();
    int methodCnt = pserviceDesc->method_count();
    std::cout << "service_name: " << service_name << std::endl;
    for (int i = 0; i < methodCnt; i++)
    {
        const google::protobuf::MethodDescriptor *pmethodDesc = pserviceDesc->method(i);
        std::string method_name = pmethodDesc->name();
        service_info.m_methodMap.insert({method_name, pmethodDesc});

        std::cout << "method_name: " << method_name << std::endl;
    }
    service_info.m_service = service;
    m_serviceMap.insert({service_name, service_info});
}

void RpcProvider::Run()
{
    // std::string ip = MprpcApplication::GetConfig().Load("rpcserverip");
    // uint16_t port = atoi(MprpcApplication::GetConfig().Load("rpcserverport").c_str());
    std::string ip = "127.0.0.1";
    uint16_t port = 8000;

    muduo::net::InetAddress address(ip, port);

    // 创建TcpServer 对象
    muduo::net::TcpServer server(&m_eventLoop, address, "RpcProvider");
    // 绑定连接回调和消息读写回调
    server.setConnectionCallback(std::bind(&RpcProvider::OnConnection, this, std::placeholders::_1));
    server.setMessageCallback(std::bind(&RpcProvider::OnMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
    server.setThreadNum(4);

    // 把当前rpc节点上要发布的服务全部注册到zk上面，让rpc client可以从zk上发现服务
    // session timeout   30s     zkclient 网络I/O线程  1/3 * timeout 时间发送ping消息
    ZkClient zkCli;
    zkCli.Start();
    // service_name为永久性节点    method_name为临时性节点
    for (auto &sp : m_serviceMap)
    {
        // /service_name   /UserServiceRpc
        std::string service_path = "/" + sp.first;
        zkCli.Create(service_path.c_str(), nullptr, 0);
        for (auto &mp : sp.second.m_methodMap)
        {
            // /service_name/method_name   /UserServiceRpc/Login 存储当前这个rpc服务节点主机的ip和port
            std::string method_path = service_path + "/" + mp.first;
            char method_path_data[128] = {0};
            sprintf(method_path_data, "%s:%d", ip.c_str(), port);
            // ZOO_EPHEMERAL表示znode是一个临时性节点
            zkCli.Create(method_path.c_str(), method_path_data, strlen(method_path_data), ZOO_EPHEMERAL);
        }
    }

    // rpc服务端准备启动，打印信息
    std::cout << "RpcProvider start service at ip:" << ip << " port:" << port << std::endl;

    server.start();
    m_eventLoop.loop();
}

void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr &conn)
{
    if (!conn->connected())
    {
        conn->shutdown();
    }
}

void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer, muduo::Timestamp)
{
    std::string recv_buf = buffer->retrieveAllAsString();
    uint32_t head_size = 0;
    recv_buf.copy((char *)&head_size, 4, 0);
    std::string rpc_header_str = recv_buf.substr(4, head_size);
    mprpc::RpcHeader rpcHeader;
    std::string service_name;
    std::string method_name;
    uint32_t args_size;

    if (rpcHeader.ParseFromString(rpc_header_str))
    {
        service_name = rpcHeader.service_name();
        method_name = rpcHeader.method_name();
        args_size = rpcHeader.args_size();
    }
    else
    {
        std::cout << "rpc_header_str " << rpc_header_str << "parse error!\n";
    }

    std::string rpc_args_str = recv_buf.substr(4 + head_size, args_size);

    auto it = m_serviceMap.find(service_name);
    if (it == m_serviceMap.end())
    {
        std::cout << service_name << "is not exist!\n";
        conn->shutdown();
        return;
    }

    auto mit = it->second.m_methodMap.find(method_name);
    if (mit == it->second.m_methodMap.end())
    {
        std::cout << method_name << "is not exist!\n";
        conn->shutdown();
        return;
    }

    google::protobuf::Service *service = it->second.m_service;
    const google::protobuf::MethodDescriptor *method = mit->second;

    // 方式一：
    google::protobuf::Message *request = service->GetRequestPrototype(method).New();
    if (!request->ParseFromString(rpc_args_str))
    {
        std::cout << "request parse error!, content: " << rpc_args_str << std::endl;
        conn->shutdown();
        return;
    }
    google::protobuf::Message *response = service->GetResponsePrototype(method).New();

    // 给下面的method方法的调用，绑定一个Closure的回调函数
    google::protobuf::Closure *done = google::protobuf::NewCallback<RpcProvider,
                                                                    const muduo::net::TcpConnectionPtr &,
                                                                    google::protobuf::Message *>(this,
                                                                                                 &RpcProvider::SendRpcResponse,
                                                                                                 conn, response);
    // mprpcController的演示代码
    //  MprpcController mprpcController;
    //  service->CallMethod(method, &mprpcController, request, response, done);
    //  if(mprpcController.Failed())
    //  {
    //      std::cout << "------------------   " << mprpcController.ErrorText() << std::endl;
    //  }
    service->CallMethod(method, NULL, request, response, done);

    // 方式二：
    // std::unique_ptr<google::protobuf::Message> request(
    //     service->GetRequestPrototype(method).New());
    // if (!request->ParseFromString(rpc_args_str))
    // {
    //     std::cerr << "request parse error!" << std::endl;
    //     return;
    // }

    // std::unique_ptr<google::protobuf::Message> response(
    //     service->GetResponsePrototype(method).New());

    // RpcDone *done = new RpcDone(conn, std::move(request), std::move(response));
    // service->CallMethod(method, nullptr, done->GetRequest(), done->GetResponse(), done);
}

void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr &conn, google::protobuf::Message *response)
{
    std::string response_str;
    if (response->SerializeToString(&response_str))
    {
        conn->send(response_str);
    }
    else
    {
        std::cerr << "serialize response failed!\n";
    }
    conn->shutdown();
}
